package com.dayu.finecomm.service;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.YamlMapFactoryBean;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.dayu.finecomm.device.IotPlatDevice;
import com.dayu.finecomm.handler.TCPHandler;
import com.dayu.finecomm.iot.IotConsole;
import com.dayu.finecomm.mongo.service.GatewayService;
import com.dayu.finecomm.mongo.service.IotPlatService;
import com.dayu.finecomm.pojo.DeviceConfig;
import com.dayu.finecomm.pojo.ServerConfig;
import com.dayu.finecomm.protocol.pojo.TStation;
import com.dayu.finecomm.source.CollectSource;
import com.dayu.framework.PieceJob;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

/**
 * @ClassName: FCService
 * @Description:TODO 通讯处理模块
 * @author: 戴卫国
 * @date: 2018年2月23日 下午2:16:22
 * 
 * @Copyright: 2018 dayu Inc. All rights reserved.
 */
@Service
public class FCService extends PieceJob {

	Logger LOGGER = LoggerFactory.getLogger(FCService.class);

	/**
	 * @Fields serverUdpPort : udp监听端口
	 */
	@Value("${service.finecomm.server.udp.port:0}")
	private int serverUdpPort = -1;

	/**
	 * udp前缀
	 */
	@Value("${service.finecomm.server.udp.prefix}")
	private String serverUdpPrefix;

	/**
	 * udp协议集合
	 */
	@Value("${service.finecomm.server.udp.protocols}")
	private String serverUdpProtocols;

	/**
	 * Tcp监听地址
	 */
	@Value("${service.finecomm.server.tcp.host}")
	private String serverTcpHost;

	/**
	 * @Fields serverTcpPort : Tcp监听端口
	 */
	@Value("${service.finecomm.server.tcp.port:-1}")
	private int serverTcpPort = -1;

	/**
	 * Tcp Json报文
	 */
	@Value("${service.finecomm.server.tcp.json:false}")
	private boolean serverTcpJson = false;

	/**
	 * Tcp 原始报文
	 */
	@Value("${service.finecomm.server.tcp.type:raw}")
	private String serverTcpType = "raw";

	/**
	 * @Fields enableUDPDistribute : UDP 转发
	 */
	@Value("${service.distribute.udp.enabled:false}")
	private boolean enableUDPDistribute = false;

	/**
	 * Udp 分发
	 */
	@Value("${service.distribute.udp.distributors}")
	private String UDPDistributors;

	/**
	 * 是否启用Mongo
	 */
	@Value("${mongodb.enable:false}")
	private boolean enableMongo;

	/**
	 * 是否启用OpcUa
	 */
	@Value("${opcua.client.enabled:false}")
	private boolean opcUaEnabled;

	/**
	 * OpcUa地址
	 */
	@Value("${opcua.server.endpoint}")
	private String opcUaEndPoint;

	/**
	 * @Fields TRANSFERS : TODO
	 */
	List<ChannelHandlerContext> TRANSFERS = new ArrayList<>();

	/**
	 * @Fields listeners : 全局的连接集合（key:ip地址 value:连接上下文）
	 */
	Map<String, ChannelHandlerContext> listeners = new ConcurrentHashMap<>();

	/**
	 * @Fields collection : TODO
	 */
	Map<String, TStation> collection = new ConcurrentHashMap<>();

	@Resource
	CollectSource stations;

	@Autowired
	@Qualifier("ymap")
	private YamlMapFactoryBean ymap;

	@Autowired
	private GatewayService gatewayService;

	@Autowired
	private IotPlatService iotplatService;

	/**
	 * @Fields futures : Netty 所有I/O执行状态
	 */
	List<ChannelFuture> futures = new ArrayList<>();

	private static List<String> clients = new ArrayList<String>();

	@Override
	public void init() throws Exception {

		// 是否开启TCP监听
		if (serverTcpPort > 0) {

			final EventLoopGroup boss = new NioEventLoopGroup();
			final EventLoopGroup workers = new NioEventLoopGroup();

			ServerBootstrap b = new ServerBootstrap();

			b.channel(NioServerSocketChannel.class);

			b.group(boss, workers);
			b.childHandler(new TCPHandler(stations, listeners));

			LOGGER.info("trace ON " + serverTcpHost + ":" + serverTcpPort);

			ChannelFuture f = null;
			if ("127.0.0.1".equals(serverTcpHost) || serverTcpHost == null) {
				f = b.bind(serverTcpPort).sync();
			} else {
				f = b.bind(serverTcpHost, serverTcpPort).sync();
			}

			ChannelFuture cf = f.channel().closeFuture();

			cf.addListener(new GenericFutureListener<Future<? super Void>>() {

				@Override
				public void operationComplete(Future<? super Void> future) throws Exception {
					boss.shutdownGracefully();
					workers.shutdownGracefully();
				}
			});

			futures.add(cf);
		}

		Set<InetSocketAddress> distributors = new HashSet<>();

		// 是否进行UDP分发
		if (enableUDPDistribute) {

			for (String s : UDPDistributors.split(",")) {

				s = s.trim();

				if (s == null || s.length() == 0) {
					LOGGER.warn("Empty string in distributors!");
				}

				if (!s.contains(":")) {
					LOGGER.warn("Illegal distributor string: " + s);
				}

				try {

					String[] ss = s.split(":", 2);

					distributors.add(new InetSocketAddress(ss[0], Integer.parseInt(ss[1])));

				} catch (Exception e) {
					LOGGER.warn("Illgal distributor string: " + s, e);
				}
			}
		}

		LOGGER.warn("loadding fimecomms 1");
		// 监听网关
		@SuppressWarnings("unchecked")
		List<Object> finecomms = (List<Object>) ymap.getObject().get("finecomms");

		LOGGER.warn("loadding fimecomms 2");
		if (finecomms != null) {

			// 监听网关集合
			List<DeviceConfig> instances = new ArrayList<>();

			for (Object o : finecomms) {

				DeviceConfig config = JSON.parseObject(JSON.toJSONString(o), DeviceConfig.class);

				instances.add(config);
			}

			// 全局UDP监听
			if (serverUdpPort > 0) {

				HashSet<String> protocols = new HashSet<>();

				if (serverUdpProtocols.length() > 0) {
					for (String s : serverUdpProtocols.split(",")) {
						s = s.trim();
						if (s.length() > 0)
							protocols.add(s);
					}
				}

				DeviceConfig config = new DeviceConfig(true, serverUdpPort, serverUdpPrefix,
						new ArrayList<>(protocols));

				instances.add(config);
			}

			// 遍历开启网关
			for (DeviceConfig config : instances) {

				if (!config.enabled || config.port <= 0) {
					continue;
				}

				config.opcUaEnabled = opcUaEnabled;
				config.opcUaEndPoint = opcUaEndPoint;

				List<String> protocols = new ArrayList<>();

				if (config.protocols != null) {
					for (String protocol : config.protocols) {
						protocols.add(protocol.toUpperCase());
					}
				}

				config.protocols = protocols;

				HashSet<String> ps = new HashSet<>();
				ps.addAll(config.protocols);

				String ty = null;
				if ("iotplat".equals(config.type)) {// iot平台驱动

					ty = "iotplat";

					IotPlatDevice iotplatDevice = new IotPlatDevice(stations, ps, config, iotplatService);

					iotplatDevice.setListeners(listeners);
					iotplatDevice.setListenType(serverTcpJson ? "json" : "raw");
					if (!"raw".equals(serverTcpType)) {
						iotplatDevice.setListenType(serverTcpType);
					}

					iotplatDevice.setDistributors(distributors);
					iotplatDevice.setPushers(config.pushers);

					IotConsole.getInstance().init(iotplatService, config.webproxy, config.refreshToken);
					IotConsole.getInstance().startHttpService("127.0.0.1", config.port, iotplatDevice);

				} else if ("slj".equals(config.type) || "heda".equals(config.type) || "wm".equals(config.type)
						|| "jh".equals(config.type) || "hla".equals(config.type) || "tcp".equals(config.type)
						|| "szxy".equals(config.type) || "jsmt".equals(config.type) || "hzsk".equals(config.type)
						|| "can".equals(config.type)) {
					// TCP驱动

					ty = "tcp";

					HdUdpService device = new HdTcpService(stations, config);

					device.getHandler().setListeners(listeners);
					device.getHandler().setListenType(serverTcpJson ? "json" : "raw");
					if (!"raw".equals(serverTcpType)) {

						device.getHandler().setListenType(serverTcpType);
					}

					device.getHandler().setDistributors(distributors);
					device.getHandler().setPushers(config.pushers);

					futures.add(device.bootstrap());

				} else if ("udp".equals(config.type) || "xg".equals(config.type) || "gdh".equals(config.type)
						|| "nbiot".equals(config.type)) {
					// UDP驱动

					ty = "udp";

					HdUdpService device = new HdUdpService(stations, config);

					device.getHandler().setListeners(listeners);
					device.getHandler().setListenType(serverTcpJson ? "json" : "raw");
					if (!"raw".equals(serverTcpType)) {

						device.getHandler().setListenType(serverTcpType);
					}

					device.getHandler().setDistributors(distributors);
					device.getHandler().setPushers(config.pushers);

					futures.add(device.bootstrap());

				} else {
					LOGGER.error("Unsupported type ");
				}

				LOGGER.info(ty + " device ON " + config.host + ":" + config.port
						+ ((config.prefix == null || config.prefix.isEmpty()) ? ""
								: (" with prefix " + config.prefix)));
			}
		}

		@SuppressWarnings("unchecked")
		List<Object> servers = (List<Object>) ymap.getObject().get("servers");

		if (servers != null) {

			List<ServerConfig> instances = new ArrayList<>();

			for (Object o : servers) {

				final ServerConfig config = JSON.parseObject(JSON.toJSONString(o), ServerConfig.class);
				instances.add(config);
			}
		}
	}

	/**
	 * <p>
	 * Title: piece
	 * </p>
	 * <p>
	 * Description: 定时查看IO读取状态，如果长时间挂起，则强制关闭
	 * </p>
	 * 
	 * @throws Exception
	 * @see com.dayu.framework.PieceJob#piece()
	 */
	@Override
	public void piece() throws Exception {

		for (ChannelFuture f : futures) {

			f.await(3650000, TimeUnit.DAYS);

			if (!f.isDone())
				f.cancel(true);
		}

	}

	public int getServerUdpPort() {
		return serverUdpPort;
	}

	public void setServerUdpPort(int serverUdpPort) {
		this.serverUdpPort = serverUdpPort;
	}

	public int getServerTcpPort() {
		return serverTcpPort;
	}

	public void setServerTcpPort(int serverTcpPort) {
		this.serverTcpPort = serverTcpPort;
	}

	public boolean isServerTcpJson() {
		return serverTcpJson;
	}

	public void setServerTcpJson(boolean serverTcpJson) {
		this.serverTcpJson = serverTcpJson;
	}

	public CollectSource getStations() {
		return stations;
	}

	public static List<String> getClients() {
		return clients;
	}

	public static void addClient(String client) {
		clients.add(client);
	}

	public static void reduceClient(String client) {
		clients.remove(client);
	}

}
